#include "RabbitMQ.h"
#include "base/time/time.h"
#include "boost/bind.hpp"
#include "trantor/utils/Logger.h"
using namespace std::literals;

RabbitMQ::RabbitMQ() : m_io_ctx(1), m_io_work(m_io_ctx), m_timer(m_io_ctx) {
    m_quit = false;
}

RabbitMQ::~RabbitMQ() {
    m_quit = true;
    m_timer.cancel();
    m_io_ctx.stop();
    if (m_thread_ptr) {
        m_thread_ptr->join();
    }
}

void RabbitMQ::init(const std::string& uri,
                    const std::string& exchange,
                    const std::string& queue,
                    const Strings& routers,
                    MQ_MSG_FUNC call_func) {
    m_routers = routers;
    m_queue_name = queue;
    m_exchange_name = exchange;
    m_mq_uri = uri;
    m_func = call_func;

    m_thread_ptr.reset(new std::thread([this]() {
        m_io_ctx.run();
    }));

    m_timer.expires_from_now(boost::posix_time::seconds(1));
    m_timer.async_wait(boost::bind(&RabbitMQ::start, this, boost::asio::placeholders::error));
}

void RabbitMQ::ack(const amqp::EnvelopePtr& msg) {
    if (m_channel) {
        m_channel->Ack(msg);
    }
}

void RabbitMQ::quit() {
    m_quit = true;
}


void RabbitMQ::start(const boost::system::error_code& e) {

    std::string tag;
    try {
        auto channel = amqp::Channel::CreateFromUri(m_mq_uri);
        m_channel = channel;

        channel->DeclareQueue(m_queue_name, false, false, false, false);

        for (size_t i = 0; i < m_routers.size(); i++) {
            channel->BindQueue(m_queue_name, m_exchange_name, m_routers[i]);
        }

        tag = channel->Consume(m_queue_name, "", true, false, false, 0);

        LOG_INFO << "mq=" << m_mq_uri << ",exchange=" << m_exchange_name << ",queue=" << m_queue_name;

        amqp::Envelope::DeliveryInfo info_now;
        int cnt = 0;
        while (!m_quit) {
            while (true) {
                amqp::EnvelopePtr msg;
                if (m_channel->ConsumeMessage(tag, msg, 10)) {
                    m_func(msg);
                    info_now = msg->GetDeliveryInfo();
                    cnt++;

                    if (cnt > 1000) {
                        break;
                    }

                } else {
                    break;
                }
            }
            if (cnt > 0) {
                m_channel->Ack(info_now, true);
                cnt = 0;
            }
        }
    } catch (const std::exception& e) {
        LOG_ERROR << "rabbit mq error queue =" << m_queue_name << ", msg=" << e.what();
        m_timer.expires_from_now(boost::posix_time::seconds(1));
        m_timer.async_wait(boost::bind(&RabbitMQ::start, this, boost::asio::placeholders::error));
    }
}
